package com.shujia.core

import org.apache.spark.rdd.RDD
import org.apache.spark.{SparkConf, SparkContext}

object Demo09Union {
  def main(args: Array[String]): Unit = {
    // 创建Spark Context
    val conf: SparkConf = new SparkConf()
    conf.setAppName("Demo09Union")
    conf.setMaster("local")

    // 控制Spark程序默认的并行度
    conf.set("spark.default.parallelism", "3")

    val sc: SparkContext = new SparkContext(conf)
    /**
     * RDD的分区数量由什么决定？
     * 1、默认情况下子RDD的分区数等于父RDD的分区数
     * 2、如果是处理HDFS的数据 第一个RDD的分区数等于“切片”的数量
     * 3、如果是shuffle类算子 则可以手动指定分区数
     * 4、可以通过Spark的参数控制分区数量 spark.default.parallelism
     *
     * shuffle类算子得到的RDD分区数判断优先级：
     * 在shuffle类算子中手动指定分区数 > spark.default.parallelism > 父RDD的数量
     *
     * 什么时候需要修改分区数？
     * 需要调整任务的并行度时可以通过改变分区数实现
     */

    val stuRDD: RDD[String] = sc.textFile("Spark/data/students.txt")

    // getNumPartitions 返回RDD中的分区数量 它不是算子 只是RDD的一个属性
    println(s"StuRDD的分区数为：${stuRDD.getNumPartitions}")

    val sampleRDD1: RDD[String] = stuRDD
      .sample(withReplacement = false, 0.1)
    println(s"sampleRDD1的分区数为：${sampleRDD1.getNumPartitions}")


    val sampleRDD2: RDD[String] = stuRDD
      .sample(withReplacement = false, 0.1)
    println(s"sampleRDD2的分区数为：${sampleRDD2.getNumPartitions}")


    // RDD与RDD之间可以通过union进行合并
    /**
     * union 转换算子
     * 可以将两个RDD进行合并
     * 两个RDD的结构必须保持一致
     * 新RDD的分区数等于两个RDD的分区数量之和
     */
    val unionRDD: RDD[String] = sampleRDD1.union(sampleRDD2)
    println(s"unionRDD的分区数为：${unionRDD.getNumPartitions}")


    //    unionRDD.foreach(println)

    // 统计性别人数
    val groupRDD: RDD[(String, Iterable[String])] = unionRDD
      .groupBy(line => line.split(",")(3))
    println(s"groupRDD的分区数为：${groupRDD.getNumPartitions}")

    val genderCntRDD: RDD[String] = groupRDD
      .map(kv => s"${kv._1},${kv._2.size}")
    println(s"genderCntRDD的分区数为：${genderCntRDD.getNumPartitions}")

    genderCntRDD.foreach(println)

    // 统计性别人数 手动给groupBy算子 指定分区数
    val groupRDD2: RDD[(String, Iterable[String])] = unionRDD
      .groupBy((line: String) => line.split(",")(3), 8)
    println(s"groupRDD2的分区数为：${groupRDD2.getNumPartitions}")


    val genderCntRDD2: RDD[String] = groupRDD2
      .map(kv => s"${kv._1},${kv._2.size}")
    println(s"genderCntRDD2的分区数为：${genderCntRDD2.getNumPartitions}")

    genderCntRDD2.foreach(println)

    // 手动改变分区数
    val newGroupRDD2: RDD[(String, Iterable[String])] = groupRDD2
      // 一般用于 减少分区数
      .coalesce(4)

    println(s"newGroupRDD2的分区数为：${newGroupRDD2.getNumPartitions}")

    val newGroupRDD3: RDD[(String, Iterable[String])] = groupRDD2
      // 一般用于 减少分区数 如果要增加分区则需将shuffle参数设为true 等同于repartition
      .coalesce(16, shuffle = true)

    println(s"newGroupRDD3的分区数为：${newGroupRDD3.getNumPartitions}")

    val newGroupRDD4: RDD[(String, Iterable[String])] = groupRDD2
      // 用于增加分区 需要产生shuffle
      .repartition(12)

    println(s"newGroupRDD4的分区数为：${newGroupRDD4.getNumPartitions}")



    while (true) {

    }
  }

}
